package org.KeyCrafter.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.redis.RedisArrayAggregator;
import io.netty.handler.codec.redis.RedisBulkStringAggregator;
import io.netty.handler.codec.redis.RedisDecoder;
import io.netty.handler.codec.redis.RedisEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import org.KeyCrafter.cmd.Command;
import org.KeyCrafter.cmd.CommandBuilder;
import org.KeyCrafter.server.aof.AofAttribute;
import org.KeyCrafter.server.aof.LoadingInfo;
import org.KeyCrafter.server.handler.CommonAcceptor;
import org.KeyCrafter.server.handler.ArgumentsAggregator;
import org.KeyCrafter.server.handler.CommandHandler;
import org.KeyCrafter.server.handler.ReplyHandler;
import org.KeyCrafter.server.session.ClientSession;

import java.io.IOException;
import java.lang.reflect.Field;
import java.nio.channels.FileChannel;
import java.nio.file.OpenOption;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import static org.KeyCrafter.server.ServerConstants.*;
import static org.KeyCrafter.server.aof.AofAttribute.KeyCrafter_AOF_OFF;
import static org.KeyCrafter.server.aof.AofAttribute.KeyCrafter_AOF_ON;
import static org.KeyCrafter.server.AppendOnlyFile.*;
import static org.KeyCrafter.util.ConfigUtils.*;
import static org.KeyCrafter.util.TimeUtil.getLRUClock;

public class KeyCrafterServer {
    private static final Logger logger = Logger.getLogger(KeyCrafterServer.class.getName());
    // 数据库
    KVDB[] db;
    // 数据库数量
    int dbNum ;
    // 命令表
    Map<String, Command> commands;
    // TCP 监听端口
    int port;
    // 一个链表，保存了所有客户端状态结构
    List<ClientSession> clients;
    // 链表，保存了所有待关闭的客户端
    List<ClientSession> clientsToClose;
    // serverCron 每秒调用的次数
    int hz;
    // 最近一次使用时钟
    int lruClock;
    // 当前客户端
    ClientSession currentClient;
    // linux时间戳（ms）
    long mstime;
    // 服务器统计数据
    ServerStatistics statistics;
    // 在执行 serverCron 时进行渐进式 rehash
    boolean activeRehashing;
    // serverCron() 函数的运行次数计数器
    int cronLoops;
    // aof 相关属性
    AofAttribute aofAttribute;
    // 后台任务线程池
    BackgroundServer backgroundServe;
    // 数据载入信息
    LoadingInfo loadingInfo;
    // 自从上次 xxx 执行以来，数据库被修改的次数
    long dirty;
    // 工作目录
    String workDir;
    String banner =
            " _  __               _____               _    \n" +
                    "| |/ /              / ____|             | |   \n" +
                    "| ' / _ __   __ _  | |     _ __ __ _ ___| | __\n" +
                    "|  < | '_ \\ / _` | | |    | '__/ _` / __| |/ /\n" +
                    "| . \\| | | | (_| | | |____| | | (_| \\__ \\   < \n" +
                    "|_|\\_\\_| |_|\\__, |  \\_____|_|  \\__,_|___/_|\\_\\\n" +
                    "             __/ |                            \n" +
                    "            |___/                             ";


    // 初始化服务器配置
    private void initServerConfig() {
        // 设置默认服务器端口号
        this.port = getIntProperty("server.port");
        // 设置数据库数量
        this.dbNum = getIntProperty("server.databases");
        //  置默认服务器频率
        this.hz = KeyCrafter_DEFAULT_HZ;
        // 设置工作目录
        this.workDir = getProperty("server.dir");
        // 初始化 AOF 配置
        this.aofAttribute = new AofAttribute();
        if (getBoolProperty("aof.enable")) {
            this.aofAttribute.setState(KeyCrafter_AOF_ON);
        } else {
            this.aofAttribute.setState(KeyCrafter_AOF_OFF);
        }

        // 设置同步策略
        String fsync = getProperty("aof.fsync");
        if ("always".equals(fsync)) {
            this.aofAttribute.setFsync(AOF_FSYNC_ALWAYS);
        } else if ("no".equals(fsync)) {
            this.aofAttribute.setFsync(AOF_FSYNC_NO);
        } else {
            this.aofAttribute.setFsync(AOF_FSYNC_EVERYSEC);
        }

        // 设置 AOF文件位置
        Path path = Paths.get(workDir,getProperty("aof.filename"));
        this.aofAttribute.setAofFilename(path.toString());


    }

    // 初始化服务器的数据结构
    private void initServer() {
        // 初始化数据库
        this.db = new DefaultKVDB[dbNum];
        for (int i = 0; i < db.length; i++) {
            db[i] = KVDB.newDefaultDb(i);
        }
        // 初始化命令表
        commands = new HashMap<>();
        for (Command command : CommandBuilder.commandTable) {
            commands.put(command.name(), command);
        }
        // 初始化客户端信息
        clients = new LinkedList<>();
        clientsToClose = new LinkedList<>();
        // 初始化统计数据
        statistics = new ServerStatistics();
        statistics.setStartTime(System.currentTimeMillis());
        // 初始化时间周期计数器
        this.cronLoops = 0;
        // 初始化线程池
        this.backgroundServe = new BackgroundServerImpl();
        // 初始化数据载入信息
        this.loadingInfo = new LoadingInfo();
        // 初始化修改信息
        this.dirty = 0;
        // 如果 AOF 持久化功能已经打开，那么打开或创建一个 AOF 文件
        if (this.aofAttribute.getState() == KeyCrafter_AOF_ON) {
            try {
                Path path = Paths.get(this.aofAttribute.getAofFilename());
                Set<OpenOption> options = new HashSet<>();
                // 以追加模式写入
                options.add(StandardOpenOption.APPEND);
                // 文件不存在就创建
                options.add(StandardOpenOption.CREATE);
                if (aofAttribute.getFsync() == AOF_FSYNC_ALWAYS) {
                    // 如果是 FSYNC_ALWAYS (总是同步) 模式
                    options.add(StandardOpenOption.DSYNC);
                }
                FileChannel fileChannel = FileChannel.open(path, options);
                this.aofAttribute.setChannel(fileChannel);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public static void printCenterAlign(String str, int len) {
        int lSpaceNum = (len - str.length()) / 2;
        for (int i = 0; i < lSpaceNum; i++) {
            System.out.print(" ");
        }
        System.out.println(str);
    }

    public static void printGodBless() {
        printCenterAlign("_ooOoo_", 150);
        printCenterAlign("o8888888o", 150);
        printCenterAlign("88\" . \"88", 150);
        printCenterAlign("(| -_- |)", 150);
        printCenterAlign("O\\ = /O", 150);
        printCenterAlign("____/'---'\\____", 150);
        printCenterAlign(".' \\\\| |// '.", 150);
        printCenterAlign("/ \\\\||| : |||// \\", 150);
        printCenterAlign("/ _||||| -:- |||||- \\", 150);
        printCenterAlign("| | \\\\\\ - /// | |", 150);
        printCenterAlign("| \\_| ''\\---/'' |_/", 150);
        printCenterAlign("\\ .-\\__ `-` ___/-. /", 150);
        printCenterAlign("___`. .' /--.--\\ `. . __", 150);
        printCenterAlign(".\"\" '< `.___\\_<|>_/___.' >'\"\".", 150);
        printCenterAlign("| | : `- \\`.;`\\ _ /`;.`/ - ` : | |", 150);
        printCenterAlign("\\ \\ `-. \\_ __\\ /__ _/ .-` / /", 150);
        printCenterAlign("======`-.____`-.___\\_____/___.-`____.-'======", 150);
        printCenterAlign("`=---='", 150);
        System.out.println();
        printCenterAlign("^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^", 150);
        printCenterAlign("  *God Bless          Never Crash* ", 150);
    }

    /**
     *
     *
     *
     */

    // 创建服务上下文
    private void createServerContext() {
        try {
            System.out.println(banner);
            System.out.println("Running at port: " + port);
//            printGodBless();
            Class<? extends ServerContext> context = ServerContext.DEFAULT.getClass();
            Field server = context.getDeclaredField("server");
            server.setAccessible(true);
            server.set(ServerContext.DEFAULT, this);
        } catch (NoSuchFieldException | IllegalAccessException e) {
            e.printStackTrace();
        }
    }

    private KeyCrafterServer() {
        initServerConfig();
        initServer();
        createServerContext();
        loadDataFromDisk();
    }

    public void run() {
        NioEventLoopGroup boss = new NioEventLoopGroup(1);
        NioEventLoopGroup worker = new NioEventLoopGroup(1);
        //boss.scheduleAtFixedRate(new ServerCron(this), 0, 1000 / this.hz, TimeUnit.MILLISECONDS);
        // 为了保证 AOF 重写的安全，用work线程执行周期任务
        worker.scheduleAtFixedRate(new ServerCron(this), 0, 1000 / this.hz, TimeUnit.MILLISECONDS);
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.channel(NioServerSocketChannel.class);
        serverBootstrap.group(boss, worker);
        // 并发安全的处理器 (如果只有一个线程的话，应该都安全)
        LoggingHandler loggingHandler = new LoggingHandler(LogLevel.DEBUG);
        CommonAcceptor commonAcceptor = new CommonAcceptor();
        CommandHandler commandHandler = new CommandHandler();
        ReplyHandler ReplyHandler = new ReplyHandler();
        serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(loggingHandler);
                ch.pipeline().addLast(commonAcceptor);
                ch.pipeline().addLast(new RedisDecoder());
                ch.pipeline().addLast(new RedisEncoder());
                ch.pipeline().addLast(new RedisBulkStringAggregator());
                ch.pipeline().addLast(new RedisArrayAggregator());
                ch.pipeline().addLast(new ArgumentsAggregator());
                ch.pipeline().addLast(commandHandler);
                ch.pipeline().addLast(ReplyHandler);
            }
        });

        try {
            System.out.println("...");
            Channel channel = serverBootstrap.bind(this.port).sync().channel();
            channel.closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }

    int LRU_CLOCK() {
        return (1000 / this.hz <= KeyCrafter_LRU_CLOCK_RESOLUTION) ?
                this.lruClock : getLRUClock();
    }

    // 从 AOF 文件中载入数据
    private void loadDataFromDisk() {
        long start = System.currentTimeMillis();
        // AOF 持久化是否打开
        if (aofAttribute.getState() == KeyCrafter_AOF_ON) {
            // 尝试载入 AOF 文件
            if (loadAppendOnlyFile(aofAttribute.getAofFilename()) == KeyCrafter_OK) {
                // 打印载入信息，并计算载入耗时长度
                //redisLog(REDIS_NOTICE,"DB loaded from append only file: %.3f seconds",(float)(ustime()-start)/1000000);
                float d = (float) (System.currentTimeMillis() - start) / 1000;
                logger.log(Level.INFO, String.format("DB loaded from append only file: %.3f seconds", d));
            }
        }
    }


    public static void main(String[] args) {
        System.out.println("starting...");
        new KeyCrafterServer().run();
    }
}
